﻿/****************************************************************************************************
	Copyright (C) 2010 RapidWebDev Organization (http://rapidwebdev.org)
	Author: Eunge, Legal Name: Jian Liu, Email: eunge.liu@RapidWebDev.org

	This program is free software: you can redistribute it and/or modify
	it under the terms of the GNU General Public License as published by
	the Free Software Foundation, either version 2 of the License, or
	(at your option) any later version.

	This program is distributed in the hope that it will be useful,
	but WITHOUT ANY WARRANTY; without even the implied warranty of
	MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
	GNU General Public License for more details.

	You should have received a copy of the GNU General Public License
	along with this program.  If not, see <http://www.gnu.org/licenses/>.
 ****************************************************************************************************/

using System;
using System.Collections.Generic;
using System.Configuration;
using System.Globalization;
using System.IO;
using System.Linq;
using System.Text;
using System.Threading;
using Lucene.Net.Analysis;
using Lucene.Net.Analysis.Snowball;
using Lucene.Net.Analysis.Standard;
using Lucene.Net.Index;
using Lucene.Net.Store;
using RapidWebDev.Common;
using RapidWebDev.SearchIndex.Contract;

namespace RapidWebDev.SearchIndex
{
	/// <summary>
	/// The service to index request documents by pulling from index message coordinator.
	/// </summary>
	public class IndexService : IIndexService
	{
        private IIndexMessageCoordinator indexMessageCoordinator;

		private static object indexQueueLocker = new object();
		private HashSet<IndexQueue> indexQueueHashSet = new HashSet<IndexQueue>();
		private Thread monitorIndexQueuesThread;
		private bool isWorking;
		private int monitorIndexQueueIntervals = 10000;

		/// <summary>
		/// Monitor index queue intervals in milliseconds, defaults to 10000.
		/// </summary>
		public int MonitorIndexQueueIntervals
		{
			get { return this.monitorIndexQueueIntervals; }
			set { this.monitorIndexQueueIntervals = value; }
		}

		/// <summary>
		/// The root directory to store all index directories.
		/// </summary>
		public string RootIndexDirectory { get; private set; }

		/// <summary>
		/// Index document handlers by indexing command.
		/// </summary>
		public IDictionary<string, IIndexDocumentHandler> Handlers { get; set; }

		/// <summary>
		/// Constructor
		/// </summary>
        public IndexService(IIndexMessageCoordinator indexMessageCoordinator)
		{
			SearchIndexConfigurationSection searchIndexConfigurationSection = ConfigurationManager.GetSection("searchIndex") as SearchIndexConfigurationSection;
			if (searchIndexConfigurationSection == null)
				throw new ConfigurationErrorsException("The configuration section \"searchIndex\" is not configured in application configuration file.");

			this.RootIndexDirectory = Kit.ToAbsolutePath(searchIndexConfigurationSection.RootIndexDirectory);
			this.Handlers = new Dictionary<string, IIndexDocumentHandler>();
            this.indexMessageCoordinator = indexMessageCoordinator;
		}

		/// <summary>
		/// Start work to index request documents.
		/// </summary>
		public IIndexService StartWork()
		{
			if (this.isWorking) return this;

			this.isWorking = true;
			this.monitorIndexQueuesThread = new Thread(new ThreadStart(this.MonitorIndexQueues));
			this.monitorIndexQueuesThread.Priority = ThreadPriority.AboveNormal;
			this.monitorIndexQueuesThread.Name = "Monitor Index Queues";
			this.monitorIndexQueuesThread.IsBackground = true;
			this.monitorIndexQueuesThread.Start();
			return this;
		}

		/// <summary>
		/// Stop indexing and abort all indexing threads.
		/// </summary>
		public void Stop()
		{
			if (this.monitorIndexQueuesThread != null && this.monitorIndexQueuesThread.ThreadState == ThreadState.Running)
			{
				// catch any exceptions on aborting the monitor thread.
				try
				{
					this.monitorIndexQueuesThread.Abort();
				}
				catch { }

				this.monitorIndexQueuesThread = null;
			}

			this.isWorking = false;
		}

		/// <summary>
		/// Monitor index queue in every {MonitorIndexQueueIntervals} seconds
		/// </summary>
		private void MonitorIndexQueues()
		{
			while (this.isWorking)
			{
				try
				{
					IEnumerable<IndexQueue> indexQueues = indexMessageCoordinator.GetIndexQueues();
					if (indexQueues.Count() > 0)
					{
						foreach (IndexQueue indexQueue in indexQueues)
						{
							lock (indexQueueLocker)
							{
								if (this.indexQueueHashSet.Contains(indexQueue)) continue;
								this.indexQueueHashSet.Add(indexQueue);
							}

							WaitCallback indexQueueProcessingThread = new WaitCallback(ProcessIndexDocMessages);
							ThreadPool.QueueUserWorkItem(indexQueueProcessingThread, indexQueue);
						}
					}

					// monitor index queue in every 10 seconds
					Thread.Sleep(this.MonitorIndexQueueIntervals);
				}
				catch (ThreadAbortException)
				{
					return;
				}
				catch (Exception exp)
				{
					Logger.Instance(this).Error(exp);
				}
			}
		}

		private void ProcessIndexDocMessages(object state)
		{
			IndexQueue indexQueue = state as IndexQueue;
	
			while (this.isWorking)
			{
				IEnumerable<IndexDocMessage> indexDocMessages = indexMessageCoordinator.Read(indexQueue);

				// terminate the indexing thread if there is no pending messages for the queue.
				if (indexDocMessages.Count() <= 0)
				{
					lock (indexQueueLocker)
					{
						if (this.indexQueueHashSet.Contains(indexQueue))
							this.indexQueueHashSet.Remove(indexQueue);
					}

					return;
				}

				this.ProcessIndexDocMessages(indexQueue, indexDocMessages);
				Thread.Sleep(25); // sleep 25ms so that the index service doesn't work so busy.
			}
		}

		private void ProcessIndexDocMessages(IndexQueue indexQueue, IEnumerable<IndexDocMessage> indexDocMessages)
		{
			string indexDirectoryPath = Path.Combine(this.RootIndexDirectory, indexQueue.ApplicationId.ToString());
			foreach (IndexDocMessage indexDocMessage in indexDocMessages)
			{
				if (!this.isWorking) return;

				try
				{
					if (!this.Handlers.ContainsKey(indexDocMessage.Command))
					{
						// log an exception.
						string errorMessage = string.Format(CultureInfo.InvariantCulture, "The index command {0} is not supported.", indexDocMessage.Command);
						indexMessageCoordinator.LogException(indexQueue, indexDocMessage.Id, errorMessage);
						continue;
					}

					IIndexDocumentHandler indexDocumentHandler = this.Handlers[indexDocMessage.Command];

					try
					{
						indexMessageCoordinator.StartIndexing(indexQueue, indexDocMessage);
						indexDocumentHandler.Process(indexDirectoryPath, indexDocMessage);
					}
					catch (Exception exp)
					{
						// log an exception.
						indexMessageCoordinator.LogException(indexQueue, indexDocMessage.Id, exp.ToString());
						continue;
					}

					// complete the document indexing.
					indexMessageCoordinator.Complete(indexQueue, indexDocMessage.Id);
				}
				catch (Exception exp)
				{
					Logger.Instance(this).Error(exp);
				}
			}
		}

		#region IDisposable Members

		/// <summary>
		/// Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources.
		/// The method stops the index service as well.
		/// </summary>
		public void Dispose()
		{
			this.Dispose(false);
		}

		/// <summary>
		/// Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources. 
		/// </summary>
		/// <param name="isDisposed">Whether the index service has been disposed.</param>
		protected virtual void Dispose(bool isDisposed)
		{
			if (isDisposed) return;
			this.Stop();
		}

		#endregion
	}
}